package addEdges;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import removeDanglingPage.ReadBytes;

import mapred.AbstractReducer;
import mapred.util.Bytes;

public class AddEdges_Reducer extends AbstractReducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>{

	@Override
	public void reduce(BytesWritable key, Iterator<BytesWritable> values,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		List<BytesWritable> link = new ArrayList<BytesWritable>();
		double pr = 0d;
		int sum = 0;
		double pagerank = 0d;
		while(values.hasNext()) {
			BytesWritable bw = values.next();
			ReadBytes b = new ReadBytes(bw);
			if(b.hasNextUrl()) {
				link.add(bw);
				sum ++;
			} else {
				pr = b.nextDouble();
				output.collect(key, bw);
			}
		}
		pagerank = pr * 0.85 / sum;
		for(BytesWritable bb : link) {
			output.collect(bb, new BytesWritable(Bytes.Double2Bytes(pagerank)));
		}
	}

//	@Override
//	public void reduce(Text key, Iterator<Text> values,
//			OutputCollector<Text, Text> output, Reporter reporter)
//			throws IOException {
//		// TODO Auto-generated method stub
//
//            List<String> link = new ArrayList<String>();
//            double pr = 0d;
//            int sum = 0;
//            double pagerank = 0d;
//            while(values.hasNext()) {
//                String s = values.next().toString().trim();
//                try {
//                    pr = Double.parseDouble(s);
//                    output.collect(key, new Text(s));
//                }
//                catch(Exception e) {
//                    link.add(s);
//                    sum ++;
//                }
//            }
//            pagerank = pr * 0.85 / sum;
//
//            for(String f : link) {
//                output.collect(new Text(f), new Text(String.valueOf(pagerank)));
//            }
//	}

}
